package com.douyu;

import com.douyu.bolt.ReportBolt;
import com.douyu.bolt.SplitSentenceBolt;
import com.douyu.bolt.WordCountBolt;
import com.douyu.spout.SentenceSpout;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;

/**
 * @Auther: xiaoyuan
 * @Date: 2019/6/19
 * @Description:
 */
public class WordCountToplogy {

    private static final String SENTENCE_SPOUT_ID = "sentence-spout";
    private static final String SPLIT_BOLT_ID = "split-bolt";
    private static final String COUNT_BOLT_ID = "count-bolt";
    private static final String REPORT_BOLT_ID = "report-bolt";
    private static final String TOPOLOGY_NAME = "word-count-topology";

    public static void main(String[] args) throws Exception{
        SentenceSpout spout = new SentenceSpout();
        SplitSentenceBolt splitSentenceBolt = new SplitSentenceBolt();
        WordCountBolt wordCountBolt = new WordCountBolt();
        ReportBolt reportBolt = new ReportBolt();
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        topologyBuilder.setSpout(SENTENCE_SPOUT_ID,spout,1);
        topologyBuilder.setBolt(SPLIT_BOLT_ID,splitSentenceBolt,1).shuffleGrouping(SENTENCE_SPOUT_ID);
        topologyBuilder.setBolt(COUNT_BOLT_ID,wordCountBolt,1).fieldsGrouping(SPLIT_BOLT_ID,new Fields("word"));
        topologyBuilder.setBolt(REPORT_BOLT_ID,reportBolt,1).globalGrouping(COUNT_BOLT_ID);
        Config config = new Config();
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology(TOPOLOGY_NAME,config,topologyBuilder.createTopology());
        Utils.sleep(10000);
        cluster.killTopology(TOPOLOGY_NAME);
        cluster.shutdown();
    }

}
